#include "threadpool.h"

const int TASK_MAX_THRESHOLD   = INT32_MAX;
const int THREAD_MAX_THRESHOLD = 1024;
const int THREAD_MAX_IDLE_TIME = 60; // 单位：秒

Result::Result( std::shared_ptr<Task> task, bool isValid ) : task_( task ), isValid_( isValid )
{
    task_->setResult( this );
}

void Result::setVal( Any any )
{
    any_ = std::move( any ); // 存储 task 的返回值
    sem_.post();             // 已经获取的任务的返回值，增加信号量资源
}

Any Result::get()
{
    if ( !isValid_ ) {
        return "";
    }
    sem_.wait(); // task 任务如果没有执行完，这里会阻塞用户的线程
    return std::move( any_ );
}

// ---------------------------------------------------------

Task::Task() : result_( nullptr ) {}

void Task::exec()
{
    if ( result_ != nullptr ) {
        result_->setVal( run() ); // 多态调用
    }
}

void Task::setResult( Result* res )
{
    result_ = res;
}

// ---------------------------------------------------------
int Thread::generateId_ = 0;

Thread::Thread( ThreadFunc func ) : func_( func ), threadId_( generateId_++ ) {}

Thread::~Thread() {}

void Thread::start()
{
    // 创建一个线程来执行一个线程函数
    std::thread t( func_, threadId_ );
    t.detach();
}

int Thread::getId() const
{
    return this->threadId_;
}

// ---------------------------------------------------------

void ThreadPool::threadFunc( int threadId )
{
    auto lastTime = std::chrono::high_resolution_clock().now();

    // 所有任务必须执行完，线程池才可以回收所有线程资源
    for ( ;; ) {
        std::shared_ptr<Task> task;
        {
            std::unique_lock<std::mutex> lock( taskQueMtx_ );

            std::cout << "tid: " << std::this_thread::get_id() << " 尝试获取任务 ..." << std::endl;

            // 每秒返回一次 区分：超时返回？有任务待执行返回？
            // 锁 + 双重判断
            while ( taskQue_.size() == 0 ) {
                // 线程池要结束，回收线程资源
                if ( !isPoolRunning_ ) {
                    threads_.erase( threadId );
                    std::cout << "threadId: " << std::this_thread::get_id() << " exit!" << std::endl;
                    exitCond_.notify_all();
                    return; // 线程函数结束，线程返回
                }

                if ( poolMode_ == PoolMode::MODE_CACHED ) {
                    // 条件变量，超时返回
                    if ( std::cv_status::timeout == notEmpty_.wait_for( lock, std::chrono::seconds( 1 ) ) ) {
                        auto now = std::chrono::high_resolution_clock().now();
                        auto dur = std::chrono::duration_cast<std::chrono::seconds>( now - lastTime );
                        if ( dur.count() >= THREAD_MAX_IDLE_TIME && curThreadSize_ > initThreadSize_ ) {
                            threads_.erase( threadId );
                            curThreadSize_--;
                            idleThreadSize_--;

                            std::cout << "threadId: " << std::this_thread::get_id() << " exit!" << std::endl;
                            return;
                        }
                    }
                } else {
                    notEmpty_.wait( lock );
                }
            }

            idleThreadSize_--;

            std::cout << "tid: " << std::this_thread::get_id() << " 获取任务成功 ..." << std::endl;

            task = taskQue_.front();
            taskQue_.pop();
            taskSize_--;

            // 如果依然有剩余任务，继续通知其它线程执行任务
            if ( !taskQue_.empty() ) {
                notEmpty_.notify_all();
            }

            idleThreadSize_++;
            lastTime = std::chrono::high_resolution_clock().now();
        }
    }
}

bool ThreadPool::checkRunningState() const
{
    return isPoolRunning_;
}

ThreadPool::ThreadPool()
    : initThreadSize_( 0 )
    , taskSize_( 0 )
    , idleThreadSize_( 0 )
    , curThreadSize_( 0 )
    , taskQueMaxThreshold_( TASK_MAX_THRESHOLD )
    , threadSizeThreshold_( TASK_MAX_THRESHOLD )
    , poolMode_( PoolMode::MODE_FIXED )
    , isPoolRunning_( false )
{}

ThreadPool::~ThreadPool()
{
    isPoolRunning_ = false;

    // 等待线程池里面所有的线程返回 有两种状态：阻塞、正在执行任务中
    std::unique_lock<std::mutex> lock( taskQueMtx_ );
    notEmpty_.notify_all();
    exitCond_.wait( lock, [&]() -> bool { return threads_.size() == 0; } );
}

void ThreadPool::setMode( PoolMode mode )
{
    if ( checkRunningState() ) {
        return;
    }
    poolMode_ = mode;
}

void ThreadPool::setTaskQueMaxThreshold( int threshold )
{
    if ( checkRunningState() ) {
        return;
    }
    threadSizeThreshold_ = threshold;
}

void ThreadPool::setThreadSizeThreshold( int threshold )
{
    if ( checkRunningState() ) {
        return;
    }
    if ( poolMode_ == PoolMode::MODE_CACHED ) {
        threadSizeThreshold_ = threshold;
    }
}

Result ThreadPool::submitTask( std::shared_ptr<Task> sp )
{
    std::unique_lock<std::mutex> lock( taskQueMtx_ );

    // 用户提交任务，最长不能阻塞超过 1s ，否则判断提交任务失败，返回
    if ( !notFull_.wait_for( lock, std::chrono::seconds( 1 ), [&]() -> bool {
             return taskQue_.size() < (size_t)taskQueMaxThreshold_;
         } ) ) {
        // 表示 notFull_ 等待 1s ，条件依然没有满足
        std::cerr << "task queue is full, submit task fail." << std::endl;
        // 现成执行完 task ，task 对象就被析构掉了
        return Result( sp, false );
    }

    // 如果有空余，把任务放入任务队列中
    taskQue_.emplace( sp );
    taskSize_++;

    notEmpty_.notify_all();

    // cached
    // 模式，任务处理比较紧急。场景：小而快的任务，需要根据任务数量和空闲线程的数量，判断是否需要创建新的线程出来
    if ( poolMode_ == PoolMode::MODE_CACHED && taskSize_ > idleThreadSize_
         && curThreadSize_ < threadSizeThreshold_ ) {
        std::cout << ">>> create new thread ..." << std::endl;

        auto ptr = std::make_unique<Thread>( std::bind( &ThreadPool::threadFunc, this, std::placeholders::_1 ) );
        int  threadId = ptr->getId();
        threads_.emplace( threadId, std::move( ptr ) );
        threads_[threadId]->start();
        curThreadSize_++;
        idleThreadSize_++;
    }

    return Result( sp );
}

void ThreadPool::start( int initThreadSize )
{
    isPoolRunning_  = true;
    initThreadSize_ = initThreadSize;
    curThreadSize_  = initThreadSize;

    // 创建线程对象
    for ( int i = 0; i < initThreadSize_; i++ ) {
        // 创建 thread 线程对象的时候，把线程函数给到 thread 线程对象
        auto ptr = std::make_unique<Thread>( std::bind( &ThreadPool::threadFunc, this, std::placeholders::_1 ) );
        int  threadId = ptr->getId();
        threads_.emplace( threadId, std::move( ptr ) );
    }

    // 启动所有线程
    for ( int i = 0; i < initThreadSize_; i++ ) {
        threads_[i]->start(); // 需要去执行一个线程函数
        idleThreadSize_++;
    }
}